package com.atguigu.flink.wordcount;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/2/23
 *
 *      在flink中，处理数据如果需要输出多个字段。
 *      多个字段可以封装为TupleN(N最多为25),但是使用起来，可读性差。
 *
 *      还可以把输出的字段封装为自定义的数据类型(POJO)
 *
 *      POJO: Plain Old  Java Object
 *                  传统的java对象。
 *              特点：  类的权限修饰符必须是public
 *                      有public的无参构造器
 *                      字段必须有公共的getter,setter
 *                      字段的类型，支持序列化
 *
 */
public class Demo5_POJO
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        //设置运行的模式
        // env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        //前提：必须是有界流
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        //DataStreamSource<String> source = env.socketTextStream("hadoop103",8888);
        DataStreamSource<String> source = env.readTextFile("data/words.txt");


        SingleOutputStreamOperator<WordCount> ds1 = source
            .flatMap(new FlatMapFunction<String, WordCount>()
            {
                @Override
                public void flatMap(String line, Collector<WordCount> out) throws Exception {
                    String[] words = line.split(" ");
                    for (String word : words) {
                        out.collect(new WordCount(word,1));
                    }
                }
            });

        ds1
            .keyBy(new KeySelector<WordCount, String>()
            {
                @Override
                public String getKey(WordCount value) throws Exception {
                    return value.getWord();
                }
            })
            //写要sum的字段
            .sum("count")
            .print();

        env.execute();

    }

    //pojo
    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class WordCount{
        private String word;
        private Integer count;
    }
}
